[FLINK-39964][state] Fix unrestorable checkpoint after CLAIM restore from native savepoint by Savonitar · Pull Request #28709 · apache/flink

Savonitar · 2026-07-09T17:14:41Z

What is the purpose of the change

The first incremental RocksDB checkpoint taken after a CLAIM-mode restore from a NATIVE savepoint reuses the savepoint's SST files. The savepoint writes those files as RelativeFileStateHandles, which are carried into the new checkpoint's shared state. The metadata serializer records only their relative path, and on restore they are re-anchored to the new checkpoint's exclusive directory. The referenced state file is therefore looked up under / while the bytes still live in /, so restoring from that checkpoint fails with a missing-file error.
The fix pushes the exclusive directory of the checkpoint being written into metadata serialization and decides, per state handle, whether the compact relative encoding is still safe: a relative handle whose file actually lives in that directory keeps it (so savepoints stay relocatable), and a foreign one (a reused savepoint SST) is persisted with its absolute path using the pre-existing FileStateHandle encoding.

Brief change log

Metadata serialization now knows the exclusive directory of the checkpoint being written: CheckpointMetadataOutputStream#getExclusiveCheckpointDir() (new, default null, overridden by FsCheckpointMetadataOutputStream) is used through a new Checkpoints.storeCheckpointMetadata(CheckpointMetadata, CheckpointMetadataOutputStream) overload, which PendingCheckpoint#finalizeCheckpoint picks up without a call-site change.
MetadataV2V3SerializerBase#canKeepRelativeEncoding(...) decides per handle: a RelativeFileStateHandle whose file lives in that directory keeps the compact relative encoding a foreign one (a reused savepoint SST) is written with its absolute path via the pre-existing FileStateHandle encoding.
Separate one-line [hotfix][tests] commit: SavepointDeepCopyTest had compared savepoint1's file listing with itself (which is incorrect) and it now reads savepoint2's directory.

Verifying this change

Serializer level: MetadataV2V3SerializerBaseTest (foreign relative handles round-trip as absolute, own handles stay relative and relocatable, null directory keeps the legacy encoding byte-for-byte) and MetadataSerializerEntryPointsTest (both entry points across all metadata versions).
End to end: ResumeCheckpointAfterClaimedNativeSavepointITCase on a real MiniCluster -> stop-with-savepoint (NATIVE) → CLAIM restore → first incremental checkpoint (asserting reused savepoint files are referenced by absolute location) → restore from that checkpoint. Plus the backend-level reproducer RocksDBClaimSavepointThenCheckpointRestoreTest.
State Processor API guard: SavepointOutputFormatSelfContainedTest and SavepointDeepCopyTest (source savepoint deleted before restoring from the copy) prove deep-copied savepoints keep no references into the source.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: yes (the checkpoint metadata serializer's write-side encoding decision changed, the wire format and metadata version are unchanged, foreign relative handles now use the pre-existing absolute FileStateHandle encoding, so the read side and older readers are unaffected)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes (that is the point of the change: the first incremental checkpoint taken after a CLAIM-mode restore from a NATIVE savepoint becomes restorable)
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)
Generated-by: Claude Code (Anthropic), Opus 4.8 / Fable.

…ry twice testSavepointDeepCopy listed savepointPath1 for both stateFiles1 and stateFiles2, so the copied-files assertion compared savepoint1's directory with itself and savepoint2's directory was never examined at all. Point stateFiles2 at savepointPath2, as the variable name, the assertion messages, and the test's documented steps always intended. The mix-up was introduced when the getFileNamesInDirectory helper was extracted (f125067); before that, both checks listed savepointPath2. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

flinkbot · 2026-07-09T17:24:28Z

CI report:

2b7a5c8 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

…from native savepoint The first incremental checkpoint after a CLAIM-mode restore from a native savepoint reuses the savepoint's SST files as RelativeFileStateHandles. The metadata serializer stores only their relative path, so on restore they are resolved against the new checkpoint's exclusive directory and can no longer be found. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

pnowojski · 2026-07-15T07:55:01Z

+    /**
+     * Stores the checkpoint metadata without knowing the checkpoint's exclusive directory: relative
+     * file references are always persisted as relative and are resolved against whatever directory
+     * the metadata is later read from. When writing the metadata of an actual checkpoint or
+     * savepoint, prefer {@link #storeCheckpointMetadata(CheckpointMetadata,
+     * CheckpointMetadataOutputStream)}.
+     *
+     * <p>The deliberately different method name keeps the context-free variants out of the {@code
+     * storeCheckpointMetadata} overload set, so a caller cannot switch between the two encodings by
+     * merely changing a variable's static type.
+     */
+    public static void storeCheckpointMetadataWithoutExclusiveDir(
            CheckpointMetadata checkpointMetadata, OutputStream out) throws IOException {

        DataOutputStream dos = new DataOutputStream(out);
-        storeCheckpointMetadata(checkpointMetadata, dos);
+        storeCheckpointMetadataWithoutExclusiveDir(checkpointMetadata, dos);
    }

+    /**
+     * Stores the checkpoint metadata into the given metadata output stream, passing the stream's
+     * exclusive checkpoint directory to the serializer so that relative file references stay
+     * self-consistent on recovery.
+     */


After reading those two java docs, I don't understand the difference between the two methods.

What is the difference between:

relative file references are always persisted as relative and are resolved against whatever directory the metadata is later read from

relative file references stay self-consistent on recovery.

?

Maybe me and this java doc is lacking explanation of what are exclusive directories?

what are exclusive directories?

Just to clarify, this PR uses pre-PR dictionary. E.g. in https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java#L934 or in https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointedStateScope.java#L31 we already have this term.
That's why I "reused" it. I'm happy to change it, if you can come up with the better option.
The exclusive directory is the per-checkpoint directory that holds _metadata and the checkpoint's non-shared state files, e.g. .../chk-100/, as opposed to the job-wide shared/ directory.

difference between the two methods.

storeCheckpointMetadataWithoutExclusiveDir(...): pre-PR behaviour, every relative handle is written relative unconditionally. It is correct behavior when the caller guarantees every referenced file is located next to the metadata (or will be located). State processor API establishes exactly this guarantee by physically copying the files

storeCheckpointMetadata(...): the serializer knows which exclusive directory it is writing into and checks every RelativeFileStateHandle against it. A handle whose file really lives there keeps the relative encoding, a handle pointing elsewhere (the reused savepoint SST from this bug) is written with its absolute path, because the relative form would be attched to chk-.../ on read and point at a file that was never there. That is what "self-consistent on recovery" meant: nothing in the written metadata resolves to a wrong location.

I will work on these java docs.

java docs updated in 70dff60

pnowojski · 2026-07-15T08:01:10Z

+                                // Must stay on the variant WITHOUT the exclusive directory. Files
+                                // retained from an existing savepoint are copied into this
+                                // directory by FileCopyFunction, but their handles still reference
+                                // the source savepoint. Writing without the exclusive directory
+                                // keeps the relative (file-name-only) encoding, which resolves
+                                // against this directory on restore and keeps the savepoint
+                                // self-contained. The exclusive-dir-aware
+                                // Checkpoints.storeCheckpointMetadata would instead fill this
+                                // savepoint's metadata with absolute paths pointing into the
+                                // source savepoint, breaking this savepoint once the source is
+                                // deleted.
+                                Checkpoints.storeCheckpointMetadataWithoutExclusiveDir(
+                                        metadata, out);


I don't understand this change and this comment doesn't help me 😓

Files retained from an existing savepoint are copied into this directory by FileCopyFunction

FileCopyFunction what's that? Who is copying what? SavepointOutputFormat doesn't have to be invovled in copy any files AFAIU.

Looking at the name, Checkpoints.storeCheckpointMetadata vs Checkpoints.storeCheckpointMetadataWithoutExclusiveDir, I don't understand how that connects to:

storeCheckpointMetadata would instead fill this savepoint's metadata with absolute paths pointing into the source savepoint

breaking this savepoint once the source is deleted.

Again I'm confused who is copying what, and why are we talking about source in this class/method.

FileCopyFunction what's that?

FileCopyFunction is a state processor API thing.

Who is copying what? SavepointOutputFormat doesn't have to be invovled in copy any files AFAIU.

SavepointWriter.write() builds a batch job with two sinks:

FileCopyFunction: copies every state file referenced by the carried-over operator states into the new savepoint directory, keeping the file name.

SavepointOutputFormat: writes the _metadata. So you're right and this class copies nothing but the other sink of the same job (FileCopyFunction) does.

Again I'm confused who is copying what, and why are we talking about source in this class/method.

The "source" is the savepoint from which one we build new one, the carried-over handles still record paths inside it, nothing rewrites them. With storeCheckpointMetadataWithoutExclusiveDir the metadata stores only file names, which resolve to the local copies on restore, so the source can be deleted.

I updated the comment significantly with this context in 3d0dda1

pnowojski · 2026-07-15T08:11:14Z

+     * <p>This is a convenience variant of {@link #serialize(CheckpointMetadata, DataOutputStream,
+     * Path)} with a {@code null} exclusive directory: relative file references keep the relative
+     * encoding unconditionally and are resolved against whatever directory the metadata is later
+     * read from, which is only correct if every referenced file actually lives in that directory.


When exclusive dir is null, and file resides outside of the _metadata's directory, shouldn't we support relative paths that are pointing away? For example files structure:

/bar/claimed-files/chk-1/_old_metadata /bar/claimed-files/chk-1/foo/1.sst /bar/checkpoints/chk-2/_metadata

AFAIU the problem in this bug is that 1.sst would be referenced in _metadata as foo/1.sst which would be incorrect. And with your fix, we would store /bar/claimed-files/chk-1/foo/1.sst. Shouldn't we instead use ../../claimed-files/chk-1/foo/1.sst? 🤔

But maybe indeed referencing claimed files by absolute path makes more sense...

pnowojski · 2026-07-15T08:19:34Z

    public void serializeOperatorStateHandleUtil(
            OperatorStateHandle stateHandle, DataOutputStream dos) throws IOException {
-        serializeOperatorStateHandle(stateHandle, dos);
+        serializeOperatorStateHandle(stateHandle, dos, null);


Why is it ok to pass null instead of a propper context in some of those calls?

These are @VisibleForTesting helpers used only from test code, they serdes standalone handles outside any checkpoint, so there is no "exclusive directory" to check relative references, that's why null is the only meaningful context there.
I added clarifications in 30c2e72

pnowojski · 2026-07-15T08:19:42Z

    public void serializeKeyedStateHandleUtil(KeyedStateHandle stateHandle, DataOutputStream dos)
            throws IOException {
-        serializeKeyedStateHandle(stateHandle, dos);
+        serializeKeyedStateHandle(stateHandle, dos, null);


ditto about null.

#28709 (comment)

pnowojski · 2026-07-15T08:20:46Z

    static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputStream dos)
            throws IOException {
+        serializeKeyedStateHandle(stateHandle, dos, null);
+    }


optional: wouldn't it be safer to just drop this variant and force everyone to expcilitly pass null or some concrete value?

I can do it. I kept the 2-arg forms to minimize the diff. I think your suggestion makes sense, but I wanted to get a feedback before doing that.

I dropped context-free methods in 30c2e72 + added justifications to remaining null

pnowojski · 2026-07-15T08:22:48Z

+     * storeCheckpointMetadata} overload set, so a caller cannot switch between the two encodings by
+     * merely changing a variable's static type.
+     */
+    public static void storeCheckpointMetadataWithoutExclusiveDir(


why do we need to keep the without excelusive dir variant? 🤔 What is/are the use cases justifying it?

I would expect that every checkpoint we are writing has it's own exclusive directory, and incorrectly referencing relative files outside of that should be explicitly forbidden?

The main use case is the state processor API's deep copy: the carried-over handles refer to outside the directory at write time, while FileCopyFunction puts a copy of every referenced file next to the new _metadata, so on read all relative references are correct. "Forbidding" would break that usage, which is correct.

…out the metadata store variants

rkhachatryan

Thanks for the fix!
The general approach LGTM, I have some minor comments, PTAL

rkhachatryan · 2026-07-15T14:04:18Z

+            // location. The trailing separator keeps a sibling directory prefix from matching.
+            final String savepointPrefix =
+                    savepointPath.endsWith("/") ? savepointPath : savepointPath + "/";
+            assertThat(sharedStateFilePaths(checkpointPath))


This might be flaky:

the 1st checkpoint might not create any SSTs

and the 2nd checkpoint might compact all the inherited SSTs out

The former can be avoided by re-taking the checkpoint; the latter can be avoided by disabling compactions.
But probably it's enough to just retry the checkpoint-restore for several iterations if this happens?

rkhachatryan · 2026-07-15T14:08:29Z

+     * unknown ({@code null}), the legacy behavior applies and the relative encoding is kept.
+     */
+    private static boolean canKeepRelativeEncoding(
+            RelativeFileStateHandle handle, @Nullable SerializationContext context) {


Could you elaborate what those legacy paths are?

I think keeping this nullable makes it a bit more error-prone without giving much value: if this change happens to be problematic there will be no way to disable it anyways

…ind SavepointOutputFormat's encoding choice [FLINK-39964][state-processor-api] Explain the deep-copy pipeline behind SavepointOutputFormat's encoding choice

…ion context explicitly

Savonitar marked this pull request as ready for review July 13, 2026 15:02

Savonitar force-pushed the fix/claim-savepoint-serialize-absolute branch from 02aeaf4 to e84e8bb Compare July 13, 2026 15:40

Savonitar changed the title ~~[FLINK-39964][checkpoint] Fix unrestorable checkpoint after CLAIM restore from native savepoint~~ [FLINK-39964][state] Fix unrestorable checkpoint after CLAIM restore from native savepoint Jul 13, 2026

rkhachatryan self-requested a review July 13, 2026 17:30

pnowojski reviewed Jul 15, 2026

View reviewed changes

[FLINK-39964][state] Explain the exclusive directory and elaborate ab…

70dff60

…out the metadata store variants

rkhachatryan reviewed Jul 15, 2026

View reviewed changes

Savonitar force-pushed the fix/claim-savepoint-serialize-absolute branch from 30c2e72 to 1776cfa Compare July 16, 2026 12:50

Savonitar added 2 commits July 16, 2026 14:59

[FLINK-39964][state-processor-api] Explain the deep-copy pipeline beh…

3d0dda1

…ind SavepointOutputFormat's encoding choice [FLINK-39964][state-processor-api] Explain the deep-copy pipeline behind SavepointOutputFormat's encoding choice

[FLINK-39964][state] Make every serializer caller pass its serializat…

2b7a5c8

…ion context explicitly

Savonitar force-pushed the fix/claim-savepoint-serialize-absolute branch from 1776cfa to 2b7a5c8 Compare July 16, 2026 12:59

Uh oh!

Conversation

Savonitar commented Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Savonitar Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Savonitar Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rkhachatryan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Savonitar commented Jul 9, 2026 •

edited

Loading

flinkbot commented Jul 9, 2026 •

edited

Loading

Savonitar Jul 15, 2026 •

edited

Loading

Savonitar Jul 15, 2026 •

edited

Loading